/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.SortedSet;
import java.util.TreeSet;

import org.slf4j.Logger;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.DCPackableString;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.server.SssServer;

public class ContIntSumReducer extends Reducer {
  private final Logger logger = SssServer.getLogger(ContIntSumReducer.class);
  private int target = 1;
  private boolean filterTop10 = false;
  private Buffer buffer = null;
  private int node = 0;
  private int slots = 1;
  private int scheduled = 0;
  @Override
  public void configure(Context context) throws Exception {
    Configuration conf = context.getConfiguration();
    this.target = conf.getInt("continuous.task.target.intervals");
    this.filterTop10 = conf.getBoolean("intsum.filter.top", false);
    this.node = conf.getInt("intsum.storage.node", 0);
    this.slots = conf.getInt("intsum.storage.slot", 1);
    this.buffer = new Buffer(target, node, slots);
    this.scheduled = 0;
    logger.info("configure target intervals {}", target);
    logger.info("output tuples will be written to storage node #{}", node);
    if (filterTop10) {
      logger.info("output top 10 records per reducer");
    }
  }
  public void reduce(Context context,
                     PackableString key, Iterable<PackableInt> values,
                     Output<DCPackableString, PackableInt> output)
    throws Exception {
    logger.trace("reduce method");
    int sum = 0;
    for (PackableInt v: values) {
      sum += v.get();
    }
    buffer.addEntry(key.get(), sum);
  }
  public void cleanup(Context context,
                      Output<DCPackableString, PackableInt> output)
    throws SssException {
    logger.trace("cleanup {}", scheduled);
    if (filterTop10) { buffer.writeTop10(output); }
    else             { buffer.writeAll(output);   }
    buffer.rotate();
    scheduled++;
  }
  private static class Buffer {
    private Map<String, Integer>[] buffer = null;
    private final int size;
    private final int node;
    private final int slots;
    private int current;
    @SuppressWarnings("unchecked")
    public Buffer(int size, int node, int slots) {
      this.buffer = new Map[size];
      this.node = node;
      this.slots = slots;
      for (int i=0; i<size; i++) {
        buffer[i] = Collections.synchronizedMap(new HashMap<String, Integer>());
      }
      this.current = 0;
      this.size = size;
    }
    public void addEntry(String key, int value) {
      buffer[current].put(key, new Integer(value));
    }
    public Set<String> keySet() {
      Set<String> ks = new HashSet<String>();
      for (int i=0; i<size; i++) {
        ks.addAll(buffer[i].keySet());
      }
      return ks;
    }
    private static class SubSet {
      private Map<Integer, List<String>> candidatesMap;
      private SortedSet<Integer> countSet;
      private final int rank;
      public SubSet(int rank) {
        this.rank = rank;
        this.countSet = new TreeSet<Integer>();
        this.candidatesMap = new HashMap<Integer, List<String>>();
      }
      public void addEntry(String key, int sum) {
        Integer intSum = new Integer(sum);
        countSet.add(intSum);
        if (countSet.size() <= rank) {
          addEntry0(key, intSum);
        }
        else {
          Integer drop = countSet.first();
          if (drop.compareTo(intSum) == 0) {
            countSet.remove(intSum);
          }
          else {
            addEntry0(key, intSum);
            countSet.remove(drop);
            candidatesMap.remove(drop);
          }
        }
      }
      private void addEntry0(String key, Integer intSum) {
        List<String> keyList = candidatesMap.get(intSum);
        if (keyList == null) {
          keyList = new ArrayList<String>();
          candidatesMap.put(intSum, keyList);
        }
        keyList.add(key);
      }
      public Set<Map.Entry<Integer, List<String>>> entrySet() {
        return candidatesMap.entrySet();
      }
    }
    public void writeTop10(Output<DCPackableString, PackableInt> output)
      throws SssException {
      SubSet subSet = new SubSet(10);
      for (String k: keySet()) {
        int sumOfSums = 0;
        for (int i=0; i<size; i++) {
          Integer sum = buffer[i].get(k);
          if (sum != null) {
            sumOfSums += sum.intValue();
          }
        }
        subSet.addEntry(k, sumOfSums);
      }
      for (Map.Entry<Integer, List<String>> e: subSet.entrySet()) {
        PackableInt v = new PackableInt(e.getKey());
        for (String s: e.getValue()) {
          output.write(new DCPackableString(s, node, s.hashCode() % slots), v);
        }
      }
    }
    public void writeAll(Output<DCPackableString, PackableInt> output)
      throws SssException {
      for (String k: keySet()) {
        int sumOfSums = 0;
        for (int i=0; i<size; i++) {
          Integer sum = buffer[i].get(k);
          if (sum != null) {
            sumOfSums += sum.intValue();
          }
        }
        output.write(new DCPackableString(k, node, k.hashCode() % slots),
                     new PackableInt(sumOfSums));
      }
    }
    public void rotate() {
      current++;
      if (current == size) {
        current = 0;
      }
      buffer[current].clear();
    }
  }
}
